package SQL_L

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.junit.Test

class WindowFun {
  //创建SparkSession
  val spark = SparkSession.builder()
    .master("local[6]")
    .appName("reader1")
    .getOrCreate()

  import org.apache.spark.sql.functions._
  import spark.implicits._

  val data = Seq(
    ("Thin", "Cell phone", 6000),
    ("Normal", "Tablet", 1500),
    ("Mini", "Tablet", 5500),
    ("Ultra thin", "Cell phone", 5000),
    ("Very thin", "Cell phone", 6000),
    ("Big", "Tablet", 2500),
    ("Bendable", "Cell phone", 3000),
    ("Foldable", "Cell phone", 3000),
    ("Pro", "Tablet", 4500),
    ("Pro2", "Tablet", 6500)
  ).toDF("product", "category", "revenue")

  /**
   * 商品销量前俩名
   */
  @Test
  def first() = {
    //定义窗口函数
    val window = Window.partitionBy('category)
      .orderBy('revenue.desc)

    //数据处理
    data.select('product, 'category, dense_rank() over(window)  as "rank")
      .where('rank <= 2)
      .show()

    //使用sql语句来完成
    data.createOrReplaceTempView("product")
    spark.sql("select product,category,revenue from " +
      " (select *,dense_rank() over (partition by category order by revenue desc) as rank from product) " +
      " where rank <= 2").show()
  }

  /**
   * 每个商品与从品类最高销量间的差值
   */
  @Test
  def second() = {
    //数据集
    val data = Seq(
      ("Thin", "Cell phone", 6000),
      ("Normal", "Tablet", 1500),
      ("Mini", "Tablet", 5500),
      ("Ultra thin", "Cell phone", 5500),
      ("Very thin", "Cell phone", 6000),
      ("Big", "Tablet", 2500),
      ("Bendable", "Cell phone", 3000),
      ("Foldable", "Cell phone", 3000),
      ("Pro", "Tablet", 4500),
      ("Pro2", "Tablet", 6500)
    ).toDF("product", "category", "revenue")

    val window = Window.partitionBy('category)
      .orderBy('revenue)
    //找到最贵的价格
    val maxPrice = max('revenue) over window
    //
    data.select('product,'category,'revenue,((max('revenue) over window) - 'revenue) as 'revenue_difference).show()

  }
}